package com.superiron.scaffold.websocket;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Websocket处理器
 */
@Component
@Slf4j
public class RealTimeMessageHandler extends TextWebSocketHandler {

    @Autowired
    private RedisTemplate redisTemplate;

    private static final String REDIS_LOCK_PREFIX = "REDIS:LOCK:";

    /*
     * 网关连接集合
     * 第一级：appId
     * 第二级：用户唯一标识 organId
     *
     * 当前节点在线session
     */
    public static Map<String, LinkedHashMap<String, WebsocketSubSession>> client_nodes = new ConcurrentHashMap<>();

    private static String redisKeyPrefix  = "WEBSOCKET:{}";

    @PostConstruct
    private void deleteRedisWebsocket() {
        Set<String> redisKeys = redisTemplate.keys(StrUtil.format(redisKeyPrefix, "*"));
        if (CollectionUtils.isNotEmpty(redisKeys)) {
            redisTemplate.delete(redisKeys);
        }
    }
    /**
     * 处理前端发送的文本信息
     * js调用websocket.send时候，会调用该方法
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        if (!session.isOpen()) {
            log.info("连接已关闭，不再处理该连接的消息！");
            return;
        }
        if (message == null || "".equals(message.getPayload())) {
            log.info(getSysDate() + "============接收到空消息，不予处理。");
        } else if (StringUtils.equals(message.getPayload(), "ping")) {
            //心跳消息过滤掉
            session.sendMessage(new TextMessage("pong"));
        } else {

            String organId = (String) session.getAttributes().get("organId");
            if (!client_nodes.containsKey(organId)) {
                session.close();
            }
            Map<String, WebsocketSubSession> session_exist_map = client_nodes.get(organId);
            if (!session_exist_map.containsKey(session.getId())) {
                session.close();
            }
            WebsocketSubSession websocketSubSession = session_exist_map.get(session.getId());
        }
    }


    /**
     * 当新连接建立的时候，被调用
     * 连接成功时候，会触发页面上onOpen方法
     *
     * @param session
     * @throws Exception
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info(getSysDate() + "============正在初始化连接：" + session.getId());
        try {
            //初始化连接，把session存储起来
            this.initUsers(session);
        } catch (Exception e) {
            log.info(getSysDate() + "============初始化连接异常-开始：" + e.getMessage());
            e.printStackTrace();
            log.info(getSysDate() + "============初始化连接异常-结束：" + e.getMessage());
        }
        log.info(getSysDate() + "============初始化连接完成：" + session.getId());
    }

    /**
     * 当连接关闭时被调用
     *
     * @param session
     * @param status
     * @throws Exception
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info(getSysDate() + "============正在关闭连接：" + session.getId() + ",isOpen:" + session.isOpen() + ";code：" + status.getCode());
        try {
            log.info("断开连接状态值" + status.getCode());
            this.removeSession(session);
        } catch (Exception e) {
            log.info(getSysDate() + "============关闭连接异常-开始：" + e.getMessage());
            e.printStackTrace();
            log.info(getSysDate() + "============关闭连接异常-结束：" + e.getMessage());
        }
        log.info(getSysDate() + "============正在关闭完成：" + session.getId() + ",isOpen:" + session.isOpen() + ";code：" + status.getCode());
    }

    /**
     * 传输错误时调用
     *
     * @param session
     * @param exception
     * @throws Exception
     */
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.error(getSysDate() + "============发生传输错误：" + session.getId() + ";session.isOpen():" + session.isOpen() + ";exception：" + exception.getMessage());
        try {
            this.removeSession(session);
        } catch (Exception e) {
            log.info(getSysDate() + "============关闭连接异常-开始：" + e.getMessage());
            e.printStackTrace();
            log.info(getSysDate() + "============关闭连接异常-结束：" + e.getMessage());
        }
    }

    //连接接入的处理方法
    private synchronized void initUsers(WebSocketSession session) {
        String organId = (String) session.getAttributes().get("organId");
        log.info("uri :" + session.getUri());
        log.info("===================================");
        if (StringUtils.isNotEmpty(organId)) {
            LinkedHashMap<String, WebsocketSubSession> session_exist_map = client_nodes.get(organId);

            if (MapUtils.isEmpty(session_exist_map)) {
                session_exist_map = new LinkedHashMap<>();
            } else {
                if (session_exist_map.keySet().size() >= 10) {
                    try {
                        for (String sessionId : session_exist_map.keySet()) {
                            WebsocketSubSession subSession = session_exist_map.get(sessionId);
                            subSession.getSession().close();
                            session_exist_map.remove(sessionId);
                            log.info("连接建立超限: organId:{},  删除之前连接:{}", organId, sessionId);
                            break;
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }

            if (session_exist_map.containsKey(session.getId())) {
                log.info("检测到相同重复连接,SN:" + organId + ",连接ID:" + session.getId() + ",准备清理失效的连接。。。");
                try {
                    WebsocketSubSession subSession = session_exist_map.get(session.getId());
                    subSession.getSession().close();
                    session_exist_map.remove(session.getId());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            WebsocketSubSession session_exist = new WebsocketSubSession();
            session_exist.setSession(session);

            session_exist_map.putIfAbsent(session.getId(), session_exist);

            client_nodes.put(organId, session_exist_map);

            log.info("连接建立: organId:{},  当前连接服务器客户端数:{}", organId, client_nodes.get(organId).keySet().size());
        }
    }

    //连接被关闭时处理集合
    private synchronized void removeSession(WebSocketSession session) {
        String organId = (String) session.getAttributes().get("organId");
        if (client_nodes.containsKey(organId)) {
            Map<String, WebsocketSubSession> session_exist_map = client_nodes.get(organId);

            WebSocketSession exist_session = session_exist_map.get(session.getId()).getSession();
            //确保是同一个session 不是同一个session则不应该进行下一步的处理
            if (exist_session != null) {
                try {
                    exist_session.close();
                } catch (IOException e) {
                    log.error("关闭session错误：{}, {}", e.getLocalizedMessage(), e);
                }
                session_exist_map.remove(session.getId());
                log.info("有一网关连接关闭！SN：" + session.getId());
            } else {
                log.info("检测到关闭session异常,程序中断处理,关闭sessionId：" + session.getId() + ",当前实际sessionId:" + exist_session.getId());
            }
        } else {
            log.info("检测到关闭session异常,程序中断处理,系统中未找到对应的session,Sn=" + session.getId());
        }
    }

    private String getSysDate() {
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
        return df.format(new Date());
    }

    /**
     *
     * @param json
     */
    public void onIotMsg(String json){
        JSONObject jsonObject = JSONObject.parseObject(json);
        String organId = jsonObject.getString("org_id");
        String redisKey = StrUtil.format(redisKeyPrefix, organId);

        if (StrUtil.isNotEmpty(organId)){

            if (MapUtils.isNotEmpty(client_nodes) && client_nodes.containsKey(organId)) {

                Map<String, WebsocketSubSession> sessionMap = client_nodes.get(organId);
                if (MapUtils.isNotEmpty(sessionMap)) {

                    for (String key : jsonObject.keySet()) {
                        if (StringUtils.equals(String.valueOf(jsonObject.get(key)), "null")) {
                            jsonObject.put(key, null);
                        }
                    }

                    for (String sessionId : sessionMap.keySet()) {
                        WebsocketSubSession websocketSubSession = sessionMap.get(sessionId);

                        if (!websocketSubSession.getPloyFlag()) {
                            synchronized (websocketSubSession) {
                                if (CollectionUtils.isNotEmpty(websocketSubSession.getDeviceIds()) && websocketSubSession.isSub(jsonObject.getString("device_id"))) {

                                    JSONObject sendMsg = null;
                                    if (CollectionUtils.isNotEmpty(websocketSubSession.getKeys())) {
                                        sendMsg = new JSONObject();
                                        sendMsg.putIfAbsent("device_id", jsonObject.getString("device_id"));
                                        sendMsg.putIfAbsent("tagid", jsonObject.getString("device_id"));
                                        sendMsg.putIfAbsent("ts", jsonObject.getLong("ts"));

                                        for (String key : websocketSubSession.getKeys()) {
                                            sendMsg.putIfAbsent(key, jsonObject.get(key));
                                        }
                                    } else {
                                        sendMsg = jsonObject;
                                    }

                                    log.info("websocket实时消息推送, 租户ID：{}, websocket_id：{}, 发送时间：{}", organId, websocketSubSession.getSession().getId(), DateUtil.formatDateTime(new Date()));
                                    try {
                                        websocketSubSession.getSession().sendMessage(new TextMessage(JSONUtil.toJsonStr(sendMsg)));
                                    } catch (IOException e) {
                                        log.error("websocket无延时发送消息失败：{}, {}", e.getLocalizedMessage(), e);
                                    }
                                }
                            }
                        }
                    }
                } else {
                    redisTemplate.delete(redisKey);
                }
            }
        }
    }
}

